package cn.chengpengper.rpc.provider.common.handler;

import cn.chengpengper.rpc.cache.result.CacheResultKey;
import cn.chengpengper.rpc.cache.result.CacheResultManager;
import cn.chengpengper.rpc.common.helper.RpcServiceHelper;
import cn.chengpengper.rpc.common.threadpool.ServerThreadPool;
import cn.chengpengper.rpc.constants.RpcConstants;
import cn.chengpengper.rpc.protocol.RpcProtocol;
import cn.chengpengper.rpc.protocol.enums.RpcStatus;
import cn.chengpengper.rpc.protocol.enums.RpcType;
import cn.chengpengper.rpc.protocol.header.RpcHeader;
import cn.chengpengper.rpc.protocol.request.RpcRequest;
import cn.chengpengper.rpc.protocol.response.RpcResponse;
import cn.chengpengper.rpc.provider.common.cache.ProviderChannelCache;
import cn.chengpengper.rpc.reflect.api.ReflectInvoker;
import cn.chengpengper.rpc.spi.loader.ExtensionLoader;
import cn.chengpengper.threadpool.ConcurrentThreadPool;
import cn.hutool.json.JSONUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Objects;

/**
 * @author chengpeng.hu
 */
@Slf4j
public class RpcProviderHandler extends SimpleChannelInboundHandler<RpcProtocol<RpcRequest>> {

    private final Map<String, Object> handlerMap;

    private ReflectInvoker reflectInvoker;

    private final boolean enableResultCache;

    private final CacheResultManager<RpcProtocol<RpcResponse>> cacheResultManager;

    private final ConcurrentThreadPool concurrentThreadPool;

    public RpcProviderHandler(String reflectType, boolean enableResultCache, int resultCacheExpire, int corePoolSize, int maximumPoolSize, Map<String, Object> handlerMap) {
        this.reflectInvoker = ExtensionLoader.getExtension(ReflectInvoker.class, reflectType);
        this.handlerMap = handlerMap;
        if (resultCacheExpire <= 0) {
            resultCacheExpire = RpcConstants.RPC_SCAN_RESULT_CACHE_EXPIRE;
        }
        this.enableResultCache = enableResultCache;
        this.cacheResultManager = CacheResultManager.getInstance(resultCacheExpire, enableResultCache);
        this.concurrentThreadPool = ConcurrentThreadPool.getInstance(corePoolSize, maximumPoolSize);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        ProviderChannelCache.add(ctx.channel());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ProviderChannelCache.remove(ctx.channel());
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        super.channelUnregistered(ctx);
        ProviderChannelCache.remove(ctx.channel());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol<RpcRequest> protocol) {
        concurrentThreadPool.submit(() -> {
            RpcProtocol<RpcResponse> responseRpcProtocol = handlerMessage(protocol, ctx.channel());
            ctx.writeAndFlush(responseRpcProtocol).addListener((ChannelFutureListener) channelFuture -> log.info("Send response for request " + protocol.getHeader().getRequestId()));
        });
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            Channel channel = ctx.channel();
            try {
                log.info("IdleStateEvent triggered,Close channel {}", channel);
                channel.close();
            } finally {
                channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
            }
        }
        super.userEventTriggered(ctx, evt);
    }

    private RpcProtocol<RpcResponse> handlerMessage(RpcProtocol<RpcRequest> protocol, Channel channel) {
        RpcProtocol<RpcResponse> responseRpcProtocol = null;
        RpcHeader header = protocol.getHeader();
        log.info("Provider receiver message:{}", JSONUtil.toJsonStr(protocol));
        if (header.getMsgType() == RpcType.HEARTBEAT_FROM_CONSUMER.getType()) {
            responseRpcProtocol = handlerHeartBeatMessage(protocol, header);
        } else if (header.getMsgType() == RpcType.HEARTBEAT_TO_PROVIDER.getType()) {
            //接收到服务消费者响应的心跳消息
            handlerHeartbeatMessageToProvider(protocol, channel);
        } else if (header.getMsgType() == RpcType.REQUEST.getType()) {
            responseRpcProtocol = handlerRequestMessageWithCache(protocol, header);
        }
        return responseRpcProtocol;
    }

    /**
     * 处理服务消费者响应的心跳消息
     */
    private void handlerHeartbeatMessageToProvider(RpcProtocol<RpcRequest> protocol, Channel channel) {
        // 收到响应信息，重置超时次数为0
        log.info("receive service consumer heartbeat message, the consumer is: {}, the heartbeat message is: {}", channel.remoteAddress(), protocol.getBody().getParameters()[0]);
        int count = ProviderChannelCache.decreaseTimes(channel);
        log.info("服务提供者收到服务消费者【{}】心跳响应，当前超时次数为【{}】", channel.remoteAddress(), count);
    }

    private RpcProtocol<RpcResponse> handlerHeartBeatMessage(RpcProtocol<RpcRequest> protocol, RpcHeader header) {
        header.setMsgType(RpcType.HEARTBEAT_TO_CONSUMER.getType().byteValue());
        RpcRequest request = protocol.getBody();
        RpcProtocol<RpcResponse> responseRpcProtocol = new RpcProtocol<>();
        RpcResponse response = new RpcResponse();
        response.setResult(RpcConstants.HEARTBEAT_PONG);
        response.setAsync(request.isAsync());
        response.setOneway(request.isOneway());
        header.setStatus(RpcStatus.SUCCESS.getCode().byteValue());
        responseRpcProtocol.setHeader(header);
        responseRpcProtocol.setBody(response);
        return responseRpcProtocol;
    }

    private RpcProtocol<RpcResponse> handlerRequestMessage(RpcProtocol<RpcRequest> protocol, RpcHeader header) {
        RpcRequest request = protocol.getBody();
        log.info("Receive request " + header.getRequestId());
        RpcProtocol<RpcResponse> responseRpcProtocol = new RpcProtocol<>();
        RpcResponse response = new RpcResponse();
        try {
            Object result = handle(request);
            response.setResult(result);
            response.setAsync(request.isAsync());
            response.setOneway(request.isOneway());
            header.setStatus(RpcStatus.SUCCESS.getCode().byteValue());
        } catch (Throwable t) {
            response.setError(t.toString());
            header.setStatus(RpcStatus.FAIL.getCode().byteValue());
            log.error("Rpc Server handle request error:", t);
        }
        responseRpcProtocol.setHeader(header);
        responseRpcProtocol.setBody(response);
        return responseRpcProtocol;
    }

    private RpcProtocol<RpcResponse> handlerRequestMessageWithCache(RpcProtocol<RpcRequest> protocol, RpcHeader header) {
        header.setMsgType(RpcType.RESPONSE.getType().byteValue());
        if (enableResultCache) {
            return handlerRequestMessageCache(protocol, header);
        }
        return handlerRequestMessage(protocol, header);
    }

    private RpcProtocol<RpcResponse> handlerRequestMessageCache(RpcProtocol<RpcRequest> protocol, RpcHeader header) {
        RpcRequest request = protocol.getBody();
        CacheResultKey cacheKey = new CacheResultKey(request.getClassName(), request.getMethodName(), request.getParameterTypes(), request.getParameters(), request.getVersion(), request.getGroup());
        RpcProtocol<RpcResponse> rpcResponseRpcProtocol = cacheResultManager.get(cacheKey);
        if (rpcResponseRpcProtocol == null) {
            rpcResponseRpcProtocol = handlerRequestMessage(protocol, header);
            // 设置保存时间
            cacheKey.setCacheTimeStamp(System.currentTimeMillis());
            cacheResultManager.put(cacheKey, rpcResponseRpcProtocol);
        }
        rpcResponseRpcProtocol.setHeader(header);
        return rpcResponseRpcProtocol;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("server caught exception", cause);
        ctx.close();
    }

    private Object handle(RpcRequest request) {
        String serviceKey = RpcServiceHelper.buildServiceKey(request.getClassName(), request.getVersion(), request.getGroup());
        Object serviceBean = handlerMap.get(serviceKey);
        if (Objects.isNull(serviceBean)) {
            throw new RuntimeException(String.format("service not exist: %s:%s", request.getClassName(), request.getMethodName()));
        }

        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        Class<?>[] parameterTypes = request.getParameterTypes();
        Object[] parameters = request.getParameters();

        log.info("RpcProviderHandler|handle|{}--{}--{}", serviceClass.getName(), methodName, parameterTypes);
        if (parameterTypes != null && parameterTypes.length > 0) {
            for (int i = 0; i < parameterTypes.length; ++i) {
                log.info(parameterTypes[i].getName());
            }
        }
        if (parameters != null && parameters.length > 0) {
            for (int i = 0; i < parameters.length; ++i) {
                log.info(parameters[i].toString());
            }
        }
        return invokeMethod(serviceBean, serviceClass, methodName, parameterTypes, parameters);
    }

    private Object invokeMethod(Object serviceBean, Class<?> serviceClass, String methodName, Class<?>[] parameterTypes, Object[] parameters) {
        try {
            return this.reflectInvoker.invokeMethod(serviceBean, serviceClass, methodName, parameterTypes, parameters);
        } catch (Throwable t) {
            log.error("invokeMethod has a exception:", t);
            throw new RuntimeException(t);
        }
    }
}
